package com.ecommerce.source;
import com.ecommerce.model.ProductData;import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import java.util.Random;
public class ProductDataSource extends RichSourceFunction<ProductData> {
    private volatile boolean running = true;
    private final Random random = new Random();
    private final String[] categories = {"女装", "男装", "美妆", "数码", "家居", "食品", "母婴", "运动"};
    private final String[] shopTypes = {"淘宝", "天猫"};
    
    @Override
    public void run(SourceContext<ProductData> ctx) throws Exception {
        long id = 1;
        while (running) {
            for (int i = 0; i < 100; i++) {
                ProductData product = new ProductData();
                product.setProductId(id++);
                product.setProductName("商品_" + random.nextInt(10000));
                product.setKeywords("关键词_" + random.nextInt(100));
                product.setShopName("店铺_" + random.nextInt(500));
                product.setPayBuyerCount(random.nextInt(10000));
                product.setVisitorCount(random.nextInt(50000));
                product.setPaymentAmount(random.nextDouble() * 100000);
                product.setGrowthRate(random.nextDouble() * 2 - 0.5); // -0.5 到 1.5
                product.setPrice(random.nextDouble() * 1000);
                product.setCategory(categories[random.nextInt(categories.length)]);
                product.setShopType(shopTypes[random.nextInt(shopTypes.length)]);
                product.setTimestamp(System.currentTimeMillis());
                
                ctx.collect(product);
            }
            Thread.sleep(1000); // 每秒生成100条数据
        }
    }
    
    @Override
    public void cancel() {
        running = false;
    }}